with shift as (
    -----根据收件收仓班次补充缺少时段
    --例如1：（08:00-17:00），则补全为（0:00-17:00）和（17:01-23:59）
    --例如2：（08:00-16:00）（16:01-22:30），则补全为（0:00-16:00）（16:01-22:30）和（22:31-23:59）
    select
        network_code,receipt_start_time_2 as receipt_start_time, receipt_end_time_2 as receipt_end_time,
        concat(receipt_start_time_2,'-',receipt_end_time_2) as hm_str
    from
        (
            select
                network_code,receipt_start_time,receipt_end_time,receipt_start_time_arr
                 ,case when ARRAY_CONTAINS(receipt_start_time_arr,receipt_start_time)!=1 and receipt_start_time='00:00' then '00:00'
                      when ARRAY_CONTAINS(receipt_start_time_arr,receipt_start_time)!=1 then from_unixtime(unix_timestamp(concat('{{ execution_date  | cst_ds }} ',receipt_start_time,':59'))+60,'HH:mm')
                  else receipt_start_time
                  end as receipt_start_time_2
                ,case when ARRAY_CONTAINS(receipt_start_time_arr,receipt_end_time)=1  then from_unixtime(unix_timestamp(concat('{{ execution_date  | cst_ds }} ',receipt_end_time,':59'))-60,'HH:mm')
                 else receipt_end_time
                 end as receipt_end_time_2
            from
                (
                SELECT
                network_code,receipt_start_time
                    ,collect_set(if(flag=1,receipt_start_time,null)) over(partition by network_code) as receipt_start_time_arr
                    ,lead(receipt_start_time,1,'23:59') over(partition by network_code order by receipt_start_time) as receipt_end_time
                from
                (
                SELECT
                network_code,receipt_start_time,flag,
                row_number() over(partition by network_code,receipt_start_time order by flag) as rk
                from
                (
                select
                network_code,flag,
                IF( flag=1 and '00:00'!= FIRST_VALUE(receipt_start_time) over(partition by network_code order by flag,receipt_start_time),'00:00',receipt_start_time)  AS receipt_start_time
                from
                (
                select network_code,receipt_start_time, 1 as flag
                from jms_dim.dim_yl_tmsnew_tms_in_warehouse_shi_base_dt where STATUS = 1 and IS_DELETE = 1 and IS_ENABLE=1 and dt ='{{ execution_date  | cst_ds }}' and receipt_start_time<receipt_end_time
                union all
                select network_code,receipt_end_time  , 2 as flag
                from jms_dim.dim_yl_tmsnew_tms_in_warehouse_shi_base_dt where STATUS = 1 and IS_DELETE = 1 and IS_ENABLE=1 and dt ='{{ execution_date  | cst_ds }}' and receipt_start_time<receipt_end_time
                union all
                select network_code,'00:00'  , 3 as flag
                from jms_dim.dim_yl_tmsnew_tms_in_warehouse_shi_base_dt where STATUS = 1 and IS_DELETE = 1 and IS_ENABLE=1 and dt ='{{ execution_date  | cst_ds }}' and receipt_start_time<receipt_end_time
                union all
                select network_code,'23:59'  , 3 as flag
                from jms_dim.dim_yl_tmsnew_tms_in_warehouse_shi_base_dt where STATUS = 1 and IS_DELETE = 1 and IS_ENABLE=1 and dt ='{{ execution_date  | cst_ds }}' and receipt_start_time<receipt_end_time
                )a
                )t
                )t2 where rk=1
                )t3 where receipt_start_time!=receipt_end_time
        )t4 where receipt_start_time_2<=receipt_end_time_2
)

insert overwrite table jms_tmp.tmp_actual_history_network_tn_dt
SELECT
    taking_code   --最早揽收网点
     ,taking_name
     ,deliver_code  --第一次派件网点
     ,deliver_name
     ,tn            --签收时间-揽收时间
     ,all_bill      --当前时段总单量
     ,split(hm_str,'-')[0] as start_hm   --揽收时分开始
     ,split(hm_str,'-')[1] as end_hm     --揽收时分结束
     ,hm_cnt        --单量
     ,cast(hm_cnt/all_bill as decimal(16,4)) as rt  --占比
     ,row_number() over(partition by  taking_code ,deliver_code,hm_str order by cast(hm_cnt/all_bill as decimal(16,4)) desc,tn) as rk --排序
    ,'{{ execution_date  | cst_ds }}' as dt
FROM
    (
        select
            taking_code   --最早揽收网点
             ,net1.name as taking_name
             ,deliver_code  --第一次派件网点
             ,net2.name as deliver_name
             ,tn --签收时间-揽收时间
             ,sum(hm_cnt) over(partition by taking_code ,deliver_code,hm_str) all_bill   --当前时段总单量
    ,hm_str
             ,hm_cnt
        from
            (
                select
                    taking_code   --最早揽收网点
                     ,deliver_code  --第一次派件网点
                     ,tn --签收时间-揽收时间
                     ,hm_str
                     ,count(1) as  hm_cnt
                from
                    (
                        SELECT
                            waybill_no
                             ,taking_code   --最早揽收网点
                             ,deliver_code  --第一次派件网点
                             ,sign_time     --入库扫描时间、签收时间、出库时间
                             ,taking_time   --揽收时间
                             ,datediff(sign_time,taking_time) as tn --签收时间-揽收时间
                             ,nvl(shift.hm_str,'00:00-23:59') as hm_str--时分
                        from
                            (
                                select
                                    waybill_no
                                     , taking_early_scansitecode  as taking_code  --最早揽收网点
                                     , deliver_early_scansitecode as deliver_code  --第一次派件网点
                                     ,coalesce(instore_early_scantime,sign_early_scantime,outstore_early_scantime) as sign_time  --入库扫描时间、签收时间、出库时间
                                     ,taking_early_scantime as taking_time --揽收时间
                                from jms_dwd.dwd_wide_unsign_summary_waybill_dt
                                where dt between date_add('{{ execution_date  | cst_ds }}',-13-14) and '{{ execution_date  | cst_ds }}'  --第一次操作时间 大多数是揽收时间 往前几天
                                  and date(coalesce(instore_early_scantime,sign_early_scantime,outstore_early_scantime)) between date_add('{{ execution_date  | cst_ds }}',-13) and '{{ execution_date  | cst_ds }}'
                                  and taking_early_scantime is not null
                                  and deliver_early_scansitecode is not null
                                  and taking_early_scansitecode is not null
                                and is_reback!=1 --剔除退件签收的
                    )t left join shift on shift.receipt_start_time<= date_format(t.taking_time,'HH:mm') and shift.receipt_end_time>= date_format(t.taking_time,'HH:mm')
                        and shift.network_code=t.taking_code
            )t2 where taking_time<sign_time --签收扫描时间要大于揽收扫描时间
        group by   taking_code  ,deliver_code ,tn,hm_str
    )t3
        left join jms_dim.dim_network_whole_massage net1 on net1.code=t3.taking_code
        left join jms_dim.dim_network_whole_massage net2 on net2.code=t3.deliver_code
    )t4
distribute by pmod(hash(rand()), 10)
;
-----取标准时效
insert  overwrite table jms_tmp.tmp_cainiao_sign_standard
SELECT
    in_from_code
     ,max(in_from_name) as in_from_name
     ,out_to_code
     ,max(out_to_name) as out_to_name
     ,out_sign_days
from
    (
        select
            in_from_code
             ,in_from_name
             ,out_to_code
             ,out_to_name
             ,out_sign_days
        from jms_dm.dm_cainiao_route_completion_data
        where dt='{{ execution_date  | cst_ds }}'
        and is_main_route !=2 --剔除航空线路
        and (out_collect_line_name!='成都-拉萨集散点1230(2班)' or out_collect_line_name is  null)
        union all
        select
              in_from_code
             ,in_from_name
             ,out_to_code
             ,out_to_name
             ,out_sign_days
        from jms_dm.dm_cainiao_route_push_old_data
        where is_main_route != 2 --剔除航空线路
        and  (out_collect_line_name!='成都-拉萨集散点1230(2班)' or out_collect_line_name is  null)
        union all
        select
              in_from_code
             ,in_from_name
             ,out_to_code
             ,out_to_name
             ,out_sign_days
        from jms_dm.dm_cainiao_supplement_tn_dt
        where dt='2023-07-31'
    )t group by in_from_code,out_to_code,out_sign_days
distribute by pmod(hash(rand()), 20)
;

-------------------------------------------------------------------------------------------
insert overwrite table jms_tmp.tmp_cainiao_history_standard_res_dt
select
    taking_code
     ,taking_name
     ,deliver_code
     ,deliver_name
     ,start_hm
     ,end_hm
     ,out_sign_days as days
     ,'{{ execution_date  | cst_ds }}' as date_time
     ,rt
     ,all_bill --当前时段总单量
     ,min(start_hm) over(partition by taking_code,deliver_code) as hm_min
     ,max(end_hm) over(partition by taking_code,deliver_code) as hm_max
     ,first_value(out_sign_days) over(partition by taking_code,deliver_code order by out_sign_days ) as allhm_top_days
     ,'{{ execution_date  | cst_ds }}' as dt
from
    (
        select  --取靠近标准时效的逻辑
            taking_code
             ,taking_name
             ,deliver_code
             ,deliver_name
             ,tn
             ,start_hm
             ,end_hm
             ,out_sign_days
             ,rt
             ,row_number() over(partition by taking_code,deliver_code,start_hm,end_hm order by diff_abs,out_sign_days) as  abs_flag
              ,all_bill
        from
            (
                select
                    t1.taking_code
                     ,t1.taking_name
                     ,t1.deliver_code
                     ,t1.deliver_name
                     ,t1.tn
                     ,t1.start_hm
                     ,t1.end_hm
                     ,t2.out_sign_days
                     ,t1.rt
                     ,abs(cast((t1.tn-t2.out_sign_days) as int)) as diff_abs
                     ,all_bill
                from
                    (select * from  jms_tmp.tmp_actual_history_network_tn_dt where dt='{{ execution_date  | cst_ds }}' and rk=1) t1
                        inner join jms_tmp.tmp_cainiao_sign_standard t2
                                   on t1.taking_code=t2.in_from_code and t1.deliver_code=t2.out_to_code

            )t3
    )t4
where abs_flag=1;
-------------------------------------------------------------------------------------------------------------

with shift_all as (
--例如1：（08:00-17:00），则补全为（0:00-07:59）\（08:00-17:00）\（17:01-23:59）
--例如2：（08:00-16:00）（16:01-22:30），则补全为 （0:00-07:59）\（08:00-16:00）（16:01-22:30）和（22:31-23:59）
    select
        taking_code,deliver_code,start_hm_2 as start_hm,end_hm_2 as end_hm
    from
        (
            select
                taking_code,deliver_code,start_hm,end_hm,start_hm_arr
                 ,case when ARRAY_CONTAINS(start_hm_arr,start_hm)!=1 and start_hm='00:00' then '00:00' 
          when ARRAY_CONTAINS(start_hm_arr,start_hm)!=1 then from_unixtime(unix_timestamp(concat('{{ execution_date  | cst_ds }} ',start_hm,':59'))+60,'HH:mm')
          else start_hm
     end as start_hm_2
    ,case when ARRAY_CONTAINS(start_hm_arr,end_hm)=1  then from_unixtime(unix_timestamp(concat('{{ execution_date  | cst_ds }} ',end_hm,':59'))-60,'HH:mm')
          else end_hm
     end as end_hm_2
            from
                (
                SELECT
                taking_code,deliver_code,start_hm
                    ,collect_set(if(flag=1,start_hm,null)) over(partition by taking_code,deliver_code) as start_hm_arr
                    ,lead(start_hm,1,'23:59') over(partition by taking_code,deliver_code order by start_hm) as end_hm
                from
                (
                SELECT
                taking_code,deliver_code,start_hm,flag,
                row_number() over(partition by taking_code,deliver_code,start_hm order by flag) as rk
                from
                (
                select taking_code,deliver_code,start_hm, 1 as flag from jms_tmp.tmp_cainiao_history_standard_res_dt where dt='{{ execution_date  | cst_ds }}'
                union all
                select taking_code,deliver_code,end_hm  , 2 as flag from jms_tmp.tmp_cainiao_history_standard_res_dt where dt='{{ execution_date  | cst_ds }}'
                union all
                select taking_code,deliver_code,'00:00'  , 3 as flag from jms_tmp.tmp_cainiao_history_standard_res_dt where dt='{{ execution_date  | cst_ds }}'
                union all
                select taking_code,deliver_code,'23:59'  , 3 as flag from jms_tmp.tmp_cainiao_history_standard_res_dt where dt='{{ execution_date  | cst_ds }}'
                )t
                )t2 where rk=1
                )t3 where start_hm!=end_hm
        )t4 where start_hm_2<=end_hm_2
)

insert overwrite table jms_dm.dm_cainiao_actual_network_tn_dt
select
    taking_code
     ,taking_name
     ,deliver_code
     ,deliver_name
     ,start_hm
     ,end_hm
     ,days
     ,date_time
     ,rt
     ,dt
from
    (   select
            taking_code
             ,taking_name
             ,deliver_code
             ,deliver_name
             ,start_hm
             ,end_hm
             ,nvl(days,jing_tn) as days --补充时段时效为空取静态最小
             ,date_time
             ,rt
             ,dt
        from
            (
                select
                    shift.taking_code
                     ,net1.name as taking_name
                     ,shift.deliver_code
                     ,net2.name as deliver_name
                     ,shift.start_hm
                     ,shift.end_hm
                     ,nvl(t5.days,t5.allhm_top_days) as days
                     ,'{{ execution_date  | cst_ds }}' as date_time
                     ,t5.rt
                     ,'{{ execution_date  | cst_ds }}' as  dt
                     --allhm_top_days 为null时  first_value()取到的是null
                     ,first_value(allhm_top_days) over(partition by shift.taking_code,shift.deliver_code order by if(allhm_top_days is null,999,allhm_top_days) ) as jing_tn
                from shift_all shift
                         left join (select * from jms_tmp.tmp_cainiao_history_standard_res_dt where dt='{{ execution_date  | cst_ds }}' )  t5
                                   on shift.taking_code=t5.taking_code and shift.deliver_code=t5.deliver_code and shift.start_hm=t5.start_hm  and shift.end_hm=t5.end_hm
                         left join jms_dim.dim_network_whole_massage net1 on net1.code=shift.taking_code
                         left join jms_dim.dim_network_whole_massage net2 on net2.code=shift.deliver_code

            )t

        union  all
        --补充历史没有的时效 但是静态路由有的时效
        select
            in_from_code as taking_code
             ,in_from_name as taking_name
             ,out_to_code as deliver_code
             ,out_to_name as deliver_name
             ,'00:00' as start_hm
             ,'23:59' as end_hm
             ,out_sign_days as days
             ,'{{ execution_date  | cst_ds }}' as date_time
             ,null as rt
             ,'{{ execution_date  | cst_ds }}' as dt
        from
            (
                SELECT
                    in_from_code
                     ,max(in_from_name) as in_from_name
                     ,out_to_code
                     ,max(out_to_name) as out_to_name
                     ,min(out_sign_days) out_sign_days
                from jms_tmp.tmp_cainiao_sign_standard
                group by in_from_code,out_to_code
            )t1
                left join
            (
                select taking_code ,deliver_code
                from  jms_tmp.tmp_actual_history_network_tn_dt
                where dt='{{ execution_date  | cst_ds }}'
                group by taking_code ,deliver_code
            )t2 on t1.in_from_code=t2.taking_code and t1.out_to_code=t2.deliver_code
        where t2.taking_code is null
    )tt
    where (deliver_code  is not null and taking_code is not null and days is not null and start_hm is not null and end_hm  is not null)
    distribute by pmod(hash(rand()), 60);